package com.wugenqiang.test.writeIntoOpentsdb;

import com.wugenqiang.test.Client.ExpectResponse;
import com.wugenqiang.test.Client.HttpClientImpl;
import com.wugenqiang.test.Client.builder.MetricBuilder;
import com.wugenqiang.test.Client.response.Response;

import java.io.IOException;

/**
 * @author wugenqiang
 * @Title: WriteIntoOpentsdb
 * @ProjectName kafka_flink_opentsdb
 * @Description: 监听数据写入opentsdb
 * @date 2018/8/29 10:35
 */
public class WriteIntoOpentsdb {

    /**
     　　* @Description: flink读取kafka数据，此处处理一条数据仅有一个数据项
     　　* @param [value]
     　　* @return void
     　　* @throws
     　　* @author wugenqiang
     　　* @date 2018/8/29 11:06
     　　*//*
    public void writeIntoOpentsdb(String value) {

        HttpClientImpl client = new HttpClientImpl("http://192.168.75.133:4242");

        MetricBuilder builder = MetricBuilder.getInstance();

        Double value2 = Double.parseDouble(value);
        builder.addMetric("com.wugenqiang.test.windpower").setDataPoint(value2)
                .addTag("tag1", "tab1value").addTag("tag2", "tab2value");

        try {
            Response response = client.pushMetrics(builder,
                    ExpectResponse.SUMMARY);
            System.out.println(response);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
*/

    /**
     　　* @Description: flink读取kafka数据，此处处理一条数据中的多个数据项
     　　* @param [value]
     　　* @return void
     　　* @throws
     　　* @author createdBy liuxiaojun editBy wugenqiang
     　　* @date 2018/8/29 12:42
     　　*/
    public void writeIntoOpentsdb(String[] value) {
        HttpClientImpl client = new HttpClientImpl("http://master:4242");

        MetricBuilder builder = MetricBuilder.getInstance();

        //时间戳
        long timeStramp = Long.parseLong(value[0]);
        //风机编号
        String fanNumber = value[1];
        //机型编号
        String model = value[2];
        //环境温度
        Double ambientTemperature =Double.valueOf(value[3]);
        //机身温度
        Double fuselageTemperature =Double.valueOf(value[4]);
        //风速
        Double windSpeed =Double.valueOf(value[5]);
        //发电量
        Double powerGeneration =Double.valueOf(value[6]);

        builder.addMetric("com.windpower.ambientTemperature").setDataPoint(timeStramp,ambientTemperature)
                .addTag("fanNumber", fanNumber).addTag("model", model);

        builder.addMetric("com.windpower.fuselageTemperature").setDataPoint(timeStramp,fuselageTemperature)
                .addTag("fanNumber", fanNumber).addTag("model", model);

        builder.addMetric("com.windpower.windSpeed").setDataPoint(timeStramp,windSpeed)
                .addTag("fanNumber", fanNumber).addTag("model", model);

        builder.addMetric("com.windpower.powerGeneration").setDataPoint(timeStramp,powerGeneration)
                .addTag("fanNumber", fanNumber).addTag("model", model);

        try {
            Response response = client.pushMetrics(builder,
                    ExpectResponse.SUMMARY);
            System.out.println(response);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
